mod gmsg;
mod umsg;

use std::collections::HashMap;
use std::time::{Duration, Instant};

use futures::{channel::mpsc, StreamExt};
use honeybee::mongodb::bson;
use honeybee::prelude::*;
use honeybee::Result;
use honeybee_proto::{client, hall};
use uuid::Uuid;

use crate::config::CONFIG;
use crate::mongo::{account::Account, role};

use super::usc::{LocalSendUser, LocalUscUserRegister, LocalUscUserUnregister, Usc};
use super::{
    stream::{SendUserMsg, UserStream},
    user_center,
};

pub struct User {
    account: String,
    uid: i64,
    create_time: i64,
    save_sender: mpsc::UnboundedSender<(i64, bson::Document)>,
    stream: Addr<UserStream>,
    #[allow(dead_code)]
    peer_addr: String,
    session_key: String,
    last_role_data: Option<bson::Document>,
    usc: Addr<Usc>,
    usc_server: Addr<TcpConnectersActor>,
    room_server: Addr<TcpConnecterActor>,
    timer1: Option<TimerCanceller>,
    heartbeat_timer: Option<TimerCanceller>,
    one_sec_interval: Option<TimerCanceller>,
    gmsg: HashMap<String, i64>,

    test_data: u32,
}

impl User {
    pub async fn new(
        account: Account,
        stream: Addr<UserStream>,
        peer_addr: String,
    ) -> Result<Self> {
        let session_key = Uuid::new_v4();
        let actor_center = ActorCenter::from_registry().await?;
        let room_server = actor_center.call(Query::new("room_server")).await?;
        let usc = actor_center.call(Query::new("usc_local")).await?;
        let usc_server = actor_center.call(Query::new("usc_server")).await?;

        let (save_sender, mut save_receiver) = mpsc::unbounded();
        spawn(async move {
            while let Some((uid, doc)) = save_receiver.next().await {
                debug!("{} save db, role:{:?}", uid, doc);
                if let Err(err) = role::update(uid, doc).await {
                    error!("uid:{} save db err:{}", uid, err);
                }
            }
        });

        Ok(Self {
            account: account.account,
            uid: account.uid,
            create_time: 0,
            save_sender,
            stream,
            peer_addr,
            session_key: session_key.to_string(),
            last_role_data: None,
            usc,
            usc_server,
            room_server,
            timer1: None,
            one_sec_interval: None,
            heartbeat_timer: None,
            gmsg: HashMap::default(),

            test_data: 0,
        })
    }

    fn send<T: 'static + MessageId + prost::Message>(&mut self, msg: T) {
        let _ = self.stream.send(SendUserMsg(msg));
    }

    fn init(&mut self, role: role::Role) -> Result<()> {
        self.last_role_data = Some(bson::to_document(&role)?);
        self.create_time = role.create_time;
        self.gmsg = role.gmsg;
        self.test_data = role.test_data;
        Ok(())
    }

    fn gen_role(&self) -> role::Role {
        role::Role {
            account: self.account.clone(),
            uid: self.uid,
            create_time: self.create_time,
            gmsg: self.gmsg.clone(),
            test_data: self.test_data,
        }
    }

    async fn save_directly(&mut self) {
        let role = self.gen_role();
        let doc = match bson::to_document(&role) {
            Ok(doc) => doc,
            Err(err) => {
                error!("{} save db save_directly err:{}", self, err);
                return;
            }
        };
        if self.last_role_data.as_ref() == Some(&doc) {
            debug!("{} not need save, no change", self);
            return;
        }
        self.last_role_data = Some(doc.clone());
        if let Err(err) = role::update(self.uid, doc).await {
            error!("{} save db save_directly err:{}", self, err);
        }
    }

    fn save(&mut self) {
        let role = self.gen_role();
        let doc = match bson::to_document(&role) {
            Ok(doc) => doc,
            Err(err) => {
                error!("{} save db err:{}", self, err);
                return;
            }
        };
        if self.last_role_data.as_ref() == Some(&doc) {
            //debug!("not need save, no change");
            return;
        }
        self.last_role_data = Some(doc.clone());
        let _ = self.save_sender.unbounded_send((self.uid, doc));
    }
}

impl Drop for User {
    fn drop(&mut self) {
        debug!("{} droped", self);
    }
}

#[async_trait::async_trait]
impl Actor for User {
    fn name(&self) -> &str {
        "User"
    }

    async fn started(&mut self, ctx: &mut Context<Self>) -> Result<()> {
        let user_center = user_center::UserCenter::from_registry().await?;
        let sequence_id = user_center
            .call(user_center::Register {
                uid: self.uid,
                session_key: self.session_key.clone(),
                user_addr: ctx.address(),
                user_stream: self.stream.clone(),
            })
            .await?;

        let uid = self.uid;
        let session_key = self.session_key.clone();
        let mut defer = Defer::new(move || {
            let _ = user_center.send(user_center::Unregister { uid, session_key });
        });

        self.usc_server
            .call_remote_hash(
                self.uid as _,
                proto::UscUserRegister {
                    uid: self.uid,
                    session_key: self.session_key.clone(),
                    sequence_id,
                    app_node: CONFIG.node.node_name.clone(),
                },
            )
            .await?;

        let role = role::load_or_create(&self.account, self.uid).await?;
        self.init(role)?;

        self.usc.send(LocalUscUserRegister {
            uid: self.uid,
            session_key: self.session_key.clone(),
            user_addr: ctx.address(),
            user_stream: self.stream.clone(),
        })?;

        self.login_process_gmsgs().await?;
        self.login_process_umsgs().await?;

        self.timer1 = Some(ctx.send_later_enable_cancel(Timer1, Duration::from_secs(10)));
        self.heartbeat_timer = Some(ctx.send_later_enable_cancel(
            Kickout("HeartBeat time out".into()),
            Duration::from_secs(10),
        ));
        //self.one_sec_interval =
        //   Some(ctx.send_interval_enable_cancel(OneSec, Duration::from_secs(1)));
        ctx.send_interval(OneSec, Duration::from_secs(1));
        ctx.send_interval(FiveSec, Duration::from_secs(5));
        ctx.send_interval(SaveDb, Duration::from_secs(60));

        self.send(hall::RspLogin {
            code: 1,
            account: self.account.clone(),
            uid: self.uid,
        });

        info!("{} user online", self);

        ctx.register_shutdown().await?;

        defer.cancel();

        Ok(())
    }

    async fn stopped(&mut self, ctx: &mut Context<Self>) {
        let _ = ctx.unregister_shutdown().await;

        self.save_directly().await;

        if let Ok(user_center) = user_center::UserCenter::from_registry().await {
            let _ = user_center.send(user_center::Unregister {
                uid: self.uid,
                session_key: self.session_key.clone(),
            });
        }
        let _ = self.usc.send(LocalUscUserUnregister {
            uid: self.uid,
            session_key: self.session_key.clone(),
        });
        if let Err(err) = self
            .usc_server
            .call_remote_hash(
                self.uid as _,
                proto::UscUserUnregister {
                    uid: self.uid,
                    session_key: self.session_key.clone(),
                },
            )
            .await
        {
            error!("{} unregister from usc err:{}", self, err);
        }
        info!("{} stoped", self);
    }
}

#[message]
pub struct Disconnected;
#[async_trait::async_trait]
impl Handler<Disconnected> for User {
    async fn handle(&mut self, ctx: &mut Context<Self>, _: Disconnected) -> Result<()> {
        ctx.stop(None);
        info!("{} disconnected", self);
        Ok(())
    }
}

impl std::fmt::Display for User {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        /*write!(
            f,
            "user account:{} uid:{},{} session_key:{}",
            self.account, self.uid, self.peer_addr, self.session_key
        )*/
        write!(f, "user account:{} uid:{}", self.account, self.uid)
    }
}

#[message]
#[derive(Clone)]
struct OneSec;
#[async_trait::async_trait]
impl Handler<OneSec> for User {
    async fn handle(&mut self, ctx: &mut Context<Self>, _: OneSec) -> Result<()> {
        //self.test_data += 1;

        self.room_server
            .send_remote(proto::UserOnlineNotifyRoom { uid: self.uid })?;

        for i in 0..100 {
            honeybee_nats::publish(
                "room",
                "app",
                proto::NatsMsgTest {
                    data: format!("this is a nats message test id:{}", i),
                },
            );
        }

        let t = Instant::now();
        match self.room_server.call_remote(proto::Add(1, 2)).await {
            Ok(result) => info!(
                "xxxxxx 1 + 2 = {} cost:{}ms",
                result,
                t.elapsed().as_millis()
            ),
            Err(err) => error!("rpc call error:{}", err),
        }
        let t = Instant::now();
        match self
            .room_server
            .call_remote_timeout(proto::GetValue, Duration::from_secs(1))
            .await
        {
            Ok(result) => info!(
                "get value = {:?} cost:{}ms",
                result,
                t.elapsed().as_millis()
            ),
            Err(err) => error!("rpc call error:{}", err),
        }
        self.room_server
            .async_call_remote(ctx.address(), proto::GetScores)?;

        let mut msg = hall::Person::default();
        for _ in 0..100 {
            msg.phone.push(hall::PhoneNumber {
                number: "hello world".to_string(),
                r#type: Some(100000),
            });
        }
        self.send(msg);
        Ok(())
    }
}

#[message]
#[derive(Clone)]
struct FiveSec;
#[async_trait::async_trait]
impl Handler<FiveSec> for User {
    async fn handle(&mut self, _ctx: &mut Context<Self>, _: FiveSec) -> Result<()> {
        self.timer1 = None;
        self.one_sec_interval = None;
        self.usc
            .send_to_user(self.uid, client::RspLocalUscAppTest {});
        self.usc
            .send_to_app_user(self.uid, proto::LocalUscAppTest {});
        /*
        self.usc_server.reliable_send_to_user(
            self.uid,
            UmsgType::UmsgTest,
            umsg_db::UmsgTest {
                order_id: 100,
                product_id: "207865".into(),
            },
        );
        */
        /*self.usc_server.reliable_send_to_all_user(
            GmsgType::GmsgTest,
            gmsg_db::GmsgTest {
                mail_msg: "you have a new mail".into(),
            },
            None,
        );*/
        Ok(())
    }
}

#[message]
#[derive(Clone)]
struct SaveDb;
#[async_trait::async_trait]
impl Handler<SaveDb> for User {
    async fn handle(&mut self, _ctx: &mut Context<Self>, _: SaveDb) -> Result<()> {
        self.save();
        Ok(())
    }
}

#[message]
#[derive(Clone)]
struct Timer1;
#[async_trait::async_trait]
impl Handler<Timer1> for User {
    async fn handle(&mut self, _ctx: &mut Context<Self>, _: Timer1) -> Result<()> {
        debug!("{} timer1 happened", self);
        Ok(())
    }
}

#[async_trait::async_trait]
impl AnyHandler<proto::Scores> for User {
    async fn handle(&mut self, _ctx: &mut Context<Self>, scores: proto::Scores) -> Result<()> {
        info!("{} async call ret:{:?}", self, scores);
        Ok(())
    }
}

#[async_trait::async_trait]
impl Handler<proto::LocalUscAppTest> for User {
    async fn handle(
        &mut self,
        _ctx: &mut Context<Self>,
        msg: proto::LocalUscAppTest,
    ) -> Result<()> {
        info!("{} receive {:?}", self, msg);
        Ok(())
    }
}

#[async_trait::async_trait]
impl AnyHandler<hall::HeartBeat> for User {
    async fn handle(&mut self, ctx: &mut Context<Self>, _: hall::HeartBeat) -> Result<()> {
        info!("receive heartbeat");
        self.send(hall::HeartBeat {});
        self.heartbeat_timer = Some(ctx.send_later_enable_cancel(
            Kickout("HeartBeat time out".into()),
            Duration::from_secs(10),
        ));
        Ok(())
    }
}

#[async_trait::async_trait]
impl Handler<proto::RoomChangeAppUserData> for User {
    async fn handle(
        &mut self,
        _ctx: &mut Context<Self>,
        msg: proto::RoomChangeAppUserData,
    ) -> Result<()> {
        info!("receive change user data:{}", msg.a);
        Ok(())
    }
}

#[message]
pub struct Kickout(pub String);
#[async_trait::async_trait]
impl Handler<Kickout> for User {
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: Kickout) -> Result<()> {
        self.send(client::RspKickout {});
        let _ = self.stream.stop(None);
        debug!("{} kickout, reason:{}", self, msg.0);
        Ok(())
    }
}

#[async_trait::async_trait]
impl Handler<proto::UscRepeatOnlineKickout> for User {
    async fn handle(
        &mut self,
        ctx: &mut Context<Self>,
        msg: proto::UscRepeatOnlineKickout,
    ) -> Result<()> {
        info!("{} receive {:?}", self, msg);
        if self.session_key != msg.session_key {
            return Ok(());
        }
        let addr = ctx.address();
        let _ = addr.send(Kickout("repeat online".into()));
        Ok(())
    }
}
